/**
 * 
 */
package com.googlecode.xcarve.heartbeat;

import io.netty.channel.Channel;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;

import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.googlecode.xcarve.common.config.CarveConstances;
import com.googlecode.xcarve.object.RPCRequest;
import com.googlecode.xcarve.transport.Client;

/**
 * @author xinyun.d
 * 2014年10月22日
 */
public class HeartBeatClientJob implements Runnable{
	private Client client;
	private int period;
	private int timeout;
	public final ConcurrentHashMap<String,Long[]> map = new ConcurrentHashMap<String,Long[]>();
	static final Logger logger = LoggerFactory.getLogger(HeartBeatClientJob.class);
	
	/**
	 * @param channles
	 * @param period
	 * @param timeout
	 */
	public HeartBeatClientJob(Client client, int period, int timeout) {
		super();
		this.client = client;
		this.period = period;
		this.timeout = timeout;
		if(timeout < period *2){
			throw new IllegalArgumentException("heartbeat timeout must < period * 2 .");
		}
	}


	@Override
	public void run() {
		try {
			Channel channel = client.getConnection();
			if(channel==null){
				return;
			}
			long now = System.currentTimeMillis();
			Attribute<Long> lastReadAttr = channel.attr(AttributeKey.<Long>valueOf(CarveConstances.LAST_READ_TIME));
			Attribute<Long> lastWriteAttr = channel.attr(AttributeKey.<Long>valueOf(CarveConstances.LAST_WRITE_TIME));
			Long lastReadTime = lastReadAttr.get();
			Long lastWriteTime = lastWriteAttr.get();
			if( lastWriteTime == null){
				sendHeartMessage(channel);
				return;
			}
			if( (lastReadTime != null && now - lastReadTime > period) || ( lastWriteTime != null && now - lastWriteTime > period) ){
				sendHeartMessage(channel);
			}
			if( lastReadTime != null && (now - lastReadTime > timeout)){
				logger.info("heartbeat message send or read timeout.");
				client.reconnection();
			} 
		} catch (Exception e) {
			logger.error("caught heartbeat exception",e);
		}
		
		
	}
	
	public void sendHeartMessage(final Channel channel) throws Exception{
		long now = System.currentTimeMillis();
		RPCRequest request = new RPCRequest();
		request.setEvent(RPCRequest.HEART_BEAT_ENVENT);
		logger.info("send heartbeat message to server. now timestamp is {}",now);
		channel.attr(AttributeKey.<Long>valueOf(CarveConstances.LAST_WRITE_TIME)).set(now);
		this.client.send(request);
	}
	
}
